package com.arxan;

import java.util.ArrayList;
import java.util.List;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
import org.apache.log4j.xml.DOMConfigurator;
import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;
import org.codehaus.jettison.json.JSONTokener;
import org.apache.hadoop.fs.FSDataInputStream;


public class HDFSFetch {
	public FileSystem fs;
	Logger log;

	/**
     * 构造函数
     *
     * @param url
     *            HDFS URL
     * @param hdfs_user
     *            HDFS 用户名
     * @return 
     */
	public HDFSFetch(String url, String hdfs_user) throws URISyntaxException, IOException {
		super();
		DOMConfigurator.configure("./log4j.xml");
		log = Logger.getLogger("com.arxan.hdfs");
		System.setProperty("HADOOP_USER_NAME", hdfs_user);
		URI uri = new URI(url);// NameNode address
		Configuration config = new Configuration();
		fs = FileSystem.get(uri, config);
	}

	/**
     * 断开HDFS连接
     *
     * @param 
     * @return 
     */
	public void finalize() {
		log.debug("hdfs.close");
		try {
			fs.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
     * 得到HDFS某路径下的所有文件和文件夹的名字
     *
     * @param path
     *            路径
     * @return 所有文件或目录的名称的字符串数组
     */
	public String[] getFileList(String path) throws IOException {

		List<String> files = new ArrayList<String>();

		Path s_path = new Path(path);
		if (fs.exists(s_path)) {
			for (FileStatus status : fs.listStatus(s_path)) {
				files.add(status.getPath().getName());
				// System.out.println("Get Name :" +
				// status.getPath().getName());
			}

		}

		return files.toArray(new String[] {});

	}

	/**
     * 向Data-Pipe发送数据，并记录处理的行数，更新计算速度
     *
     * @param posturl
     *            发送请求的 URL
     * @param p2pcompany
     * 			  公司编号
     * @param datatype
     * 			  数据类型，包括transcationdata,transfer,scatteredinvest,updatestatus,financing
     * @param proctargetfile
     * 			  要处理的文件的名称
     * @param calcSpeed
     * 			  用于计算处理速度的线程
     * @param dbsave
     *            记录已经处理的行数的数据库对象
     * @return 总共处理的数据条数
     */
	public int ProcessTarget(String posturl, String p2pcompany, String datatype, String proctargetfile,
			CalcSpeed calcSpeed, int oldstep, DBSaveStep dbsave) throws IllegalArgumentException, IOException {
		int proccount = 0;
		FSDataInputStream inputStream = fs.open(new Path(proctargetfile));
		
		inputStream.skipBytes(oldstep);
		BufferedReader br = new BufferedReader(new InputStreamReader(inputStream));
		//br.skip(oldstep);
		
		int oncesize = 0;
		boolean procfinished = false;
		int newstep = oldstep;
		while (!procfinished) {
			String postData = "[";
			String lineTxt = null;
			int readline = 0;
			oncesize = 0;
			while (readline < 500 && ((lineTxt = br.readLine()) != null) ) {
				if (lineTxt == null || lineTxt.equals(""))
					continue;
				
				postData += lineTxt + ",";
				readline++;
				oncesize+= lineTxt.getBytes("UTF-8").length;
				newstep += lineTxt.getBytes("UTF-8").length+2;
			}
			if (readline > 0) {
				postData = StringUtils.substringBeforeLast(postData, ",");
				postData += "]";
				String trueposturl = posturl + datatype;
				//log.info("first send data : " + postData.substring(0,100));
				String response = DataPipeClient.sendPost(trueposturl, postData);
				log.debug(" step " + newstep + " Response: " + response  );
				//TODO parse response
				//{"recode":"0000","message":"","product_reg_code":""}
				JSONTokener jsonTokener = new JSONTokener(response);
				JSONObject respJSONObject;
				try {  
					respJSONObject = (JSONObject) jsonTokener.nextValue();  
		            String recode = respJSONObject.getString("recode");  
		            if(!recode.equals("0000")){
		            	log.error("post url failed" + response);
		            	System.exit(-1);
		            }
		              
		        } catch (JSONException e) {  
		            e.printStackTrace();  
		            log.error("post url failed:"+ response + " json error:" + e.getMessage());
	            	System.exit(-1);
		        }  
				
			} else {
				procfinished = true;
				log.info("finished: " + proctargetfile);
			}
			log.debug("file " + proctargetfile + " new step: " + newstep);
			dbsave.updateFileStep(proctargetfile, newstep);
			proccount += readline;
			calcSpeed.processedItem(readline,oncesize);
		}

		br.close();
		inputStream.close();
		return proccount;
	}
}
